package com.byron.media.server.ffmpeg;

import com.byron.media.server.config.MediaServerConfig;
import com.byron.media.server.utils.FrameInfo;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;

import java.util.Queue;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.atomic.AtomicBoolean;

import static org.bytedeco.javacpp.avcodec.AVPacket;
import static org.bytedeco.javacpp.avcodec.av_packet_unref;
import static org.bytedeco.javacpp.avformat.*;
import static org.bytedeco.javacpp.avutil.*;

@Slf4j
public class AnyReceiver implements Runnable {

    // 不支持UDP, RTP
    // 支持RTSP, HTTP
    private int maxMills;            // 当已经收到最大毫秒数为max时候，暂停拉取

    private int minMills;            // 当已经收到最小毫秒数为min时候，就继续拉取

    private String url;

    private boolean running;

    private AVPacket packet;

    private AVFormatContext format_ctx;

    private int video_stream_index;
    
    private int audio_stream_index;

    private FrameListener frameListener;

    private int width = 1280;      // 编码视频大小

    private int height = 720;     // 编码视频大小

    private AnyToH264 anyToH264;

    private AnyToAAC anyToAAC;

    private final Object playLock = new Object();          // 用于暂停转录

    private AtomicBoolean pause = new AtomicBoolean(false);

    private Queue<PtsFrameInfo> ptsFrameQueue = new ConcurrentLinkedQueue<>();

    private long startTime;     // 开始播放的时间

    private long timeToReceiveMax = Long.MAX_VALUE;          // 到达拉取最大值时的时间

    private AtomicBoolean pullingStop = new AtomicBoolean(false);                // 是否停止了拉取

    private long firstPts = -1;          // 用于记录第一个包的pts 和 dts

    private long firstDts = -1;

    public AnyReceiver(MediaServerConfig config, int width, int height, String url, FrameListener frameListener) {
        this.minMills = config.getMinMills();
        this.maxMills = config.getMaxMills();
        this.url = url;
        this.width = width;
        this.height = height;
        this.anyToH264 = new AnyToH264();
        this.anyToAAC = new AnyToAAC();
        this.frameListener = frameListener;
    }

    /**
     * 暂停编码
     */
    public void pause(){
        pause.set(true);
    }

    /**
     * 继续编码
     */
    public void play(){
        if(!pause.getAndSet(false)){
            return;
        }
        synchronized (playLock) {
            playLock.notify();
        }
    }

    @Override
    public void run() {
        this.startTime = System.currentTimeMillis();
        try {
            initFfmpeg();
            anyToH264.initFfmpeg(format_ctx, width, height);

            // 解码出来之后推入到一个队列中
            anyToH264.setFrameListener(new FrameListener() {
                @Override
                public void onFrame(FrameInfo frameInfo) {

                    if(firstPts == -1){
                        firstPts = frameInfo.getPts();
                    }

                    if(firstDts == -1){
                        firstDts = frameInfo.getDts();
                    }

                    // 视频统一为H264
                    if(frameInfo.isVideo()){
//                        long millsInVideo = frameInfo.getPts() / 90;
//                        long millsInVideo = frameInfo.getPts()  / 5;
//                        long millsInVideo = frameInfo.getPts()  / 50;
//                        long millsInVideo = (frameInfo.getPts() - firstPts) * 1000 / 50000;

                        // 视频时间戳转毫秒
//                        long millsInVideo = (frameInfo.getPts() - firstPts) * anyToH264.getVideoNum() / anyToH264.getVideoDen() * 1000;
                        long millsInVideo = (frameInfo.getPts() - firstPts) / 90000 ;
                        long mills = millsInVideo + startTime;

                        byte[] data = new byte[frameInfo.getLength() - frameInfo.getOffset()];
                        System.arraycopy(frameInfo.getFrameData(), frameInfo.getOffset(), data, 0, data.length);
                        ptsFrameQueue.offer(new PtsFrameInfo(mills, new FrameInfo(data, 0, data.length, frameInfo.getPts(), frameInfo.getDts(), true, frameInfo.getDuration())));

                        long now = System.currentTimeMillis();
                        if(mills > (now + maxMills)){

                            timeToReceiveMax = mills;

                            // TODO 触发暂停拉取（限制网速）
                            pullingStop.set(true);

                        } else {
                            pullingStop.set(false);
                        }

                    // 音频统一为AAC
                    } else {
                        // TODO 音频推入队列
                    }
                }
            });

            // TODO AAC 处理
            int ret = 0;
//            long bitRate = format_ctx.bit_rate();
            while(running){

                long now = System.currentTimeMillis();
                PtsFrameInfo ptsFrame = ptsFrameQueue.peek();
                if(ptsFrame != null){

//                    if(now + minMills < timeToReceiveMax && pullingStop.get()){
//                        timeToReceiveMax = Long.MAX_VALUE;
//
//                        // TODO 当包的时间小于最小时间，则继续播放
//                        pullingStop.set(false);
//                    }

                    if(ptsFrame.getMillis() < now){
                        ptsFrame = ptsFrameQueue.poll();
                        // 输出该Frame

                        if(frameListener != null){
                            frameListener.onFrame(ptsFrame.frameInfo);
                        }
                    }
                }

                // 暂停处理
                while (pause.get()){
                    synchronized (playLock) {
                        try {
                            playLock.wait();
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                    }
                }

                if(pullingStop.get()){

                    if(now + minMills > timeToReceiveMax && pullingStop.get()){
                        // TODO 当包的时间小于最小时间，则继续播放
                        pullingStop.set(false);
                    }

                    Thread.sleep(5);
                    continue;
                }

                ret = av_read_frame(format_ctx, packet);
                if (ret < 0) {
                    log.info(String.format("error or end of file: %d\n", ret));
                    running = false;
                    close();
                    continue;
                }

                if (packet.stream_index() == video_stream_index) {
                    int length = packet.size();
                    byte[] h264 = new byte[length];
                    packet.position(0);
                    packet.data().get(h264, 0, length);
                    anyToH264.handle(h264, packet.pts(), packet.dts(), packet.duration());
                }

                if (packet.stream_index() == audio_stream_index) {
                    int length = packet.size();
                    byte[] aac = new byte[length];
                    packet.data().position(0).get(aac, 0, length);
                    anyToAAC.handle(aac, packet.pts(), packet.dts());
                }

                av_packet_unref(packet);
            }
//            releaseFfmpeg();
            anyToH264.releaseFfmpeg();
        } catch (Exception ex) {
            log.error(ex.toString(), ex);
        }
    }

    /**
     * 开始线程
     */
    public void start() {
        running = true;
        new Thread(this).start();
    }

    /**
     * 转换结果监听器
     * @param frameListener
     */
    public void setFrameListener(FrameListener frameListener) {
        this.frameListener = frameListener;
    }

    public void close() {
        running = false;
    }

    private int releaseFfmpeg(){
        if(format_ctx != null){
            avformat_free_context(format_ctx);
            format_ctx.setNull();
            format_ctx = null;
        }
        return 0;
    }

    private int initFfmpeg(){
        avformat_network_init();
        // Allocate an AVFormatContext
        format_ctx = avformat_alloc_context();

        // open rtsp: Open an input stream and read the header. The codecs are not opened
        int ret = -1;
        ret = avformat_open_input(format_ctx, url, null, null);
        if (ret != 0) {
            log.info(String.format("fail to open url: %s, return value: %d\n", url, ret));
            return -1;
        }

        // Read packets of a media file to get stream information
        ret = avformat_find_stream_info(format_ctx, (AVDictionary) null);
        if ( ret < 0) {
            log.info(String.format("fail to get stream information: %d\n", ret));
            return -1;
        }

        // audio/video stream index
        video_stream_index = -1;
        audio_stream_index = -1;
        log.info(String.format("Number of elements in AVFormatContext.streams: %d\n", format_ctx.nb_streams()));
        for (int i = 0; i < format_ctx.nb_streams(); ++i) {
		    AVStream stream = format_ctx.streams(i);
            log.info(String.format("type of the encoded data: %d\n", stream.codecpar().codec_id()));
            if (stream.codecpar().codec_type() == AVMEDIA_TYPE_VIDEO) {
                video_stream_index = i;
                log.info(String.format("dimensions of the video frame in pixels: width: %d, height: %d, pixel format: %d\n",
                        stream.codecpar().width(), stream.codecpar().height(), stream.codecpar().format()));
            } else if (stream.codecpar().codec_type() == AVMEDIA_TYPE_AUDIO) {
                audio_stream_index = i;
                log.info(String.format("audio sample format: %d\n", stream.codecpar().format()));
            }
        }

        if (video_stream_index == -1) {
            log.info(String.format("no video stream\n"));
            return -1;
        }

        if (audio_stream_index == -1) {
            log.info(String.format("no audio stream\n"));
        }

        packet = new AVPacket();

        return 0;
    }

    @Data
    @AllArgsConstructor
    public class PtsFrameInfo {

        private long millis;                // 播放时间

        private FrameInfo frameInfo;        // 源帧数据

    }

}
